/*
 *
 * -------------------------------------------------------------------------------------------------
 *                            COPYRIGHT(C) BOCO HEALTH 2016
 * -------------------------------------------------------------------------------------------------
 *
 * =================================================================================================
 *  *HISTORY
 *
 *      Tag                Date            Author           Description
 * =================================================================================================
 *      created            2017/1/17      lizhineng        create
 * =================================================================================================
 *
 *
 */

package com.boco.base.rxbus;


import com.orhanobut.logger.Logger;

import java.util.HashMap;

import rx.Observable;
import rx.Subscription;
import rx.android.schedulers.AndroidSchedulers;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.schedulers.Schedulers;
import rx.subjects.ReplaySubject;
import rx.subjects.SerializedSubject;
import rx.subjects.Subject;
import rx.subscriptions.CompositeSubscription;

/**
 * 创建人： lizhineng
 * 创建时间：2016/09/20
 * 邮箱：lizhineng@boco.com.cn
 */
public class RxBus {
    private static volatile RxBus mRxBus;

    //主题
    private final Subject<Object, Object> mSubject;

    private HashMap<String, CompositeSubscription> mSubscriptionHashMap;

    // PublishSubject只会把在订阅发生的时间点之后，来自原始Observable的数据发射给观察者
    // BehaviorSubject会把在订阅发生的时间点之前最后一个事件缓存起来，来自原始Observable的数据发射给观察者
    // ReplaySubject会把在订阅发生的时间点之前的所有事件缓存起来，来自原始Observable的数据发射给观察者
    private RxBus() {
        mSubject = new SerializedSubject<>(ReplaySubject.create());
    }

    /**
     * 获取单例RxBus
     *
     * @return
     */
    public static RxBus getInstance() {
        if (mRxBus == null) {
            synchronized (RxBus.class) {
                if (mRxBus == null) {
                    mRxBus = new RxBus();
                }
            }
        }
        return mRxBus;
    }

    /**
     * 发送一个单一类型的事件
     *
     * @param o 事件数据
     */
    public void post(Object o) {
        mSubject.onNext(o);
    }


    /**
     * 发送一个新事件,根据code进行分发
     *
     * @param code 事件code
     * @param o    事件数据
     */
    public void post(int code, Object o) {
        mSubject.onNext(new Message(code, o));
    }

    /**
     * 根据传递的eventType类型返回特定类型(eventType)的被观察者
     *
     * @param eventType 事件类型
     * @param <T>
     * @return
     */
    public <T> Observable<T> toObservable(Class<T> eventType) {
        return mSubject.ofType(eventType);
    }

    /**
     * 根据传递的code和eventType类型返回特定类型(eventType)的被观察者
     *
     * @param code     事件code
     * @param evenType 事件类型
     * @param <T>
     * @return
     */
    public <T> Observable<T> toObservable(final int code, final Class<T> evenType) {
        return mSubject.ofType(Message.class)
                .filter(new Func1<Message, Boolean>() {

                    @Override
                    public Boolean call(Message message) {
                        return message.getCode() == code
                                && evenType.isInstance(message.getObject());
                    }
                })
                .map(new Func1<Message, Object>() {
                    @Override
                    public Object call(Message message) {
                        return message.getObject();
                    }
                })
                .cast(evenType);
    }

    /**
     * 是否已有观察者订阅
     *
     * @return
     */
    public boolean hasObservers() {
        return mSubject.hasObservers();
    }

    /**
     * 单一事件默认的订阅方法
     *
     * @param eventType 事件类型
     * @param next
     * @param error
     * @param <T>
     * @return
     */
    public <T> Subscription doSubscribe(Class<T> eventType, Action1<T> next, Action1<Throwable>
            error) {
        return toObservable(eventType)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(next, error);
    }

    /**
     * 根据传递的code和eventType类型返回特定类型(eventType)的被观察者的默认订阅事件
     *
     * @param code      事件code
     * @param eventType 事件类型
     * @param next
     * @param error
     * @param <T>
     * @return
     */
    public <T> Subscription doSubscribe(int code, Class<T> eventType, Action1<T> next,
                                        Action1<Throwable> error) {
        return toObservable(code, eventType)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(next, error);
    }

    /**
     * 保存订阅后的subscription 第二个参数只传一个表示单一eventType 传两个表示多个订阅依据code区分
     *
     * @param subscription 订阅事件
     * @param o            1.传入当前类就行 2.code
     */
    public void addSubscription(Subscription subscription, Object... o) {
        if (mSubscriptionHashMap == null) {
            mSubscriptionHashMap = new HashMap<>();
        }
        String key;
        if (o.length == 1) key = o[0].getClass().getName();
        else key = o[0].getClass().getName() + "_" + o[1];
        if (mSubscriptionHashMap.get(key) != null) {
            mSubscriptionHashMap.get(key).add(subscription);
        } else {
            CompositeSubscription compositeSubscription = new CompositeSubscription();
            compositeSubscription.add(subscription);
            mSubscriptionHashMap.put(key, compositeSubscription);
        }
    }

    /**
     * 取消订阅
     *
     * @param o 1.传入当前类就行 2.code
     */
    public void unSubscribe(Object... o) {
        if (mSubscriptionHashMap == null) {
            return;
        }
        String key;
        if (o.length == 1) key = o[0].getClass().getName();
        else key = o[0].getClass().getName() + "_" + o[1];
        if (!mSubscriptionHashMap.containsKey(key)) {
            return;
        }
        if (mSubscriptionHashMap.get(key) != null) {
            mSubscriptionHashMap.get(key).unsubscribe();
        }

        mSubscriptionHashMap.remove(key);
        Logger.e(key+"：已取消订阅");
    }

}
